package com.flink.cus;

import java.time.Duration;

import com.flink.cus.operators.FMap1;
import com.flink.cus.operators.Sink1;
import com.flink.cus.operators.Sink2;
import com.flink.cus.operators.SinkA;
import com.flink.cus.wdopers.MyTrigger;
import com.flink.cus.wdopers.stockTsWmarkPerdAssigner;
import com.flink.cus.wdopers.stockTsWmarkPerdAssigner2;
import com.flink.cus.wdopers.stockTsWmarkPuncAssigner;
import com.flink.util.ConfTool;
import com.flink.cus.winfun.MyReduceSum;

import org.apache.flink.api.common.ExecutionConfig.GlobalJobParameters;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class WindowDemo {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT,"8081-8089");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

        ConfTool.ReadConfig("flink.conf");  
        GlobalJobParameters gp = ParameterTool.fromPropertiesFile(ConfTool.getConfFile());        
        env.getConfig().setGlobalJobParameters(gp);

        DataStreamSource<String> source = env.socketTextStream("172.100.0.1", 8818); 
        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(1000);
        
        WatermarkStrategy<Tuple3<Long, Float, Long>> wstg = 
        WatermarkStrategy.<Tuple3<Long, Float, Long>>forBoundedOutOfOrderness(Duration.ofMillis(100))
        .withTimestampAssigner((event, timestamp) -> event.f0).withIdleness(Duration.ofMinutes(1));
        
        AllWindowedStream<Tuple3<Long, Float, Long>, TimeWindow> wd = source         
            .flatMap(new FMap1()).uid("fmap01").name("读出报价")
            .assignTimestampsAndWatermarks(wstg)
            //.assignTimestampsAndWatermarks(new stockTsWmarkPerdAssigner2(Time.milliseconds(100)))  //before 1.12
            .windowAll(SlidingEventTimeWindows.of(Time.seconds(1), Time.seconds(1)))
            .trigger(EventTimeTrigger.create())
            .evictor(TimeEvictor.of(Time.seconds(1),true))
            .allowedLateness(Time.seconds(1))
        ;        

        SinkA sink1 = new SinkA(); sink1.SetName("Min");
        SinkA sink2 = new SinkA(); sink2.SetName("Max");
        SinkA sink3 = new SinkA(); sink3.SetName("Sum");
        wd.min(1).addSink(sink1).uid("sink01").name("打印1");
        wd.max(1).addSink(sink2).uid("sink02").name("打印2");
        wd.reduce(new MyReduceSum()).addSink(sink3).uid("sink03").name("打印3");
        //wd.aggregate(new AggregationFunction<Tuple3<Long, Float, Long>>());

        env.execute("window demo1");
    }

}
